package com.shujia.indicators.core;

import com.shujia.indicators.utils.SparkUtils;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

import java.text.SimpleDateFormat;
import java.time.LocalDate;
import java.time.ZoneId;
import java.util.Calendar;
import java.util.Date;

public class MarketingRemindEtl {
    public static void main(String[] args) {
        SparkSession sparkSession = SparkUtils.initSession();

        LocalDate localDate = LocalDate.of(2019, 11, 30);

        Date now = Date.from(localDate.atStartOfDay(ZoneId.systemDefault()).toInstant());

        Calendar calendar = Calendar.getInstance();
        SimpleDateFormat foramt = new SimpleDateFormat("yyyy-MM-dd");
        calendar.setTime(now);
        calendar.add(Calendar.DATE,-7);
        Date sevenDayBefore = calendar.getTime();
        String format = foramt.format(sevenDayBefore);

        String freeCountSql ="select date_format(create_time,'yyyy-MM-dd') as day," +
                "count(member_id) as free_count " +
                "from bigdata005_user_portraits_dim.dim_t_coupon_member where coupon_id =1 and coupon_channel=2 and create_time >= '"+format +"'" +
                " group by date_format(create_time,'yyyy-MM-dd')";

        Dataset<Row> freeCountdataset = sparkSession.sql(freeCountSql);

        freeCountdataset.write().format("jdbc")
                .mode(SaveMode.Append)
                .option("url","jdbc:mysql://localhost:3306/bigdata005?useSSL=false&useUnicode=true&characterEncoding=UTF-8")
                .option("dbtable","free_count")
                .option("user","bigdata005")
                .option("password","ShuJia666777...")
                .option("driver","com.mysql.cj.jdbc.Driver")
                .save();


        String couponCountSql ="select date_format(create_time,'yyyy-MM-dd') as day," +
                "count(member_id) as coupon_count " +
                "from bigdata005_user_portraits_dim.dim_t_coupon_member where coupon_id !=1 and create_time >= '"+format +"'" +
                " group by date_format(create_time,'yyyy-MM-dd')";

        Dataset<Row> couponCountDataset = sparkSession.sql(couponCountSql);

        couponCountDataset.write().format("jdbc")
                .mode(SaveMode.Append)
                .option("url","jdbc:mysql://localhost:3306/bigdata005?useSSL=false&useUnicode=true&characterEncoding=UTF-8")
                .option("dbtable","coupon_count")
                .option("user","bigdata005")
                .option("password","ShuJia666777...")
                .option("driver","com.mysql.cj.jdbc.Driver")
                .save();

    }
}
